package com.tc.vms.utils;

import com.tc.vms.exception.VmsClientErrorCode;
import com.tc.vms.exception.VmsClientException;
import com.tc.vms.log.VmsClientLogger;
import lombok.extern.slf4j.Slf4j;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkMarshallingError;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.I0Itec.zkclient.serialize.ZkSerializer;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;
import java.util.regex.Pattern;

/**
 * zookeeper access layer
 * Created by yonghua.zhang on 2015/12/2.
 */
@Slf4j
public class ZkUtils {
    public final static Logger logger = VmsClientLogger.getLog();
    public final static Pattern vmsClientFilterPat = Pattern.compile("([\\d*]+\\.){3}[\\d*]+");
    public final static String bootstrapCfgCenter = "GD6-cfgcenter-ZK-001.idc.vip.com:2181,GD6-cfgcenter-ZK-002.idc.vip.com:2181,GD6-cfgcenter-ZK-003.idc.vip.com:2181";
    public final static String bootstrapCfgCenterEnv = "VIP_CFGCENTER_ZK_CONNECTION";
    public final static String bootstrapCfgPartition = "VIP_CFGCENTER_PARTITION";
    public final static String vmsPartition = "VIP_VMS_PARTITION";
    public final static String zkChannelPath = "channel";
    public final static String zkQueuePath = "queue";
    public final static String zkProducerPath = "producer";
    public final static String zkIllegalPath = "illegal";
    public final static String zkClusterPath = "cluster";
    public final static String vmsChannelMetadataTopic = "vms.inboundName";
    public final static String vmsChannelSubQueues = "sharding.queues";
    public final static String vmsIsQueueSharding = "is.queue.sharding";
    public final static String vmsRabbitmqShards = "rmq.queue.shards";

    //zk command dispatcher
    public final static String vmsLoggerSwitcher = "vms.log.switcher";
    public final static String vmsSuspendSwitcher = "vms.suspend.switcher";
    public final static String vmsTpsLimit = "vms.tps.limit";
    public final static String vmsClientFilter = "vms.gray.release.filter";

    public final static String vmsMetadataCluster = "vms.clusterId";
    public final static String vmsMetadataUser = "username";
    public final static String vmsMetadataPasswd = "password";
    public final static String vmsMetadataVhost = "vhost";

    public final static String vmsQueueMetadataGroup = "vms.outboundName";
    public final static String vmsQueueMetadataBindings = "vms.bindings";

    public final static String vmsClusterMetadataBroker = "brokerServer";
    public final static String vmsClusterMetadataCfgsvr = "configServer";
    public final static String vmsClusterMetadataBrokerType = "brokerServer.product";
    public final static String vmsClusterMetadataCfgsvrType = "configServer.product";
    public final static String vmsCharEncoding = "UTF-8";

    private final static Set<String> vmsCommonAttr = new HashSet<String>();
    private final static Set<String> vmsControlAttr = new HashSet<String>();

    static {
        vmsCommonAttr.add(vmsChannelMetadataTopic);
        vmsCommonAttr.add(vmsChannelSubQueues);
        vmsCommonAttr.add(vmsIsQueueSharding);
        vmsCommonAttr.add(vmsRabbitmqShards);
        vmsCommonAttr.add(vmsMetadataCluster);
        vmsCommonAttr.add(vmsMetadataUser);
        vmsCommonAttr.add(vmsMetadataPasswd);
        vmsCommonAttr.add(vmsMetadataVhost);
        vmsCommonAttr.add(vmsQueueMetadataGroup);
        vmsCommonAttr.add(vmsQueueMetadataBindings);
        vmsCommonAttr.add(vmsClusterMetadataBroker);
        vmsCommonAttr.add(vmsClusterMetadataCfgsvr);
        vmsCommonAttr.add(vmsClusterMetadataBrokerType);
        vmsCommonAttr.add(vmsClusterMetadataCfgsvrType);

        vmsControlAttr.add(vmsLoggerSwitcher);
        vmsControlAttr.add(vmsSuspendSwitcher);
        vmsControlAttr.add(vmsTpsLimit);
    }

    public final static Properties getVmsCommonProps(Properties props) {
        Properties rtnProps = new Properties();
        for (String key : props.stringPropertyNames()) {
            if (vmsCommonAttr.contains(key)) {
                rtnProps.put(key, props.getProperty(key));
            }
        }
        return rtnProps;
    }

    public final static Properties getVmsControlProps(Properties props) {
        Properties rtnProps = new Properties();
        for (String key : props.stringPropertyNames()) {
            if (vmsControlAttr.contains(key)) {
                rtnProps.put(key, props.getProperty(key));
            }
        }
        return rtnProps;
    }

    public final static String getVmsQueueBindings(Properties props) {
        return props.getProperty(vmsQueueMetadataBindings);
    }

    public final static String getVmsVhost(Properties props) {
        return props.getProperty(vmsMetadataVhost, "default");
    }

    public final static boolean getVmsIsSuspend(Properties props) {
        try {
            return Boolean.valueOf(props.getProperty(vmsSuspendSwitcher, "false"));
        } catch (Exception e) {
        }
        return false;
    }

    public final static boolean getVmsLoggerSwitcher(Properties props) {
        try {
            return Boolean.valueOf(props.getProperty(vmsLoggerSwitcher, "false"));
        } catch (Exception e) {
        }
        return true;
    }

    public final static int getVmsTpsLimit(Properties props) {
        try {
            return Integer.parseInt(props.getProperty(vmsTpsLimit, "0"));
        } catch (Exception e) {
        }
        return 0;
    }

    public final static boolean hasVmsClientFilter(Properties props) {
        String rtnStr = props.getProperty(vmsClientFilter);
        return rtnStr != null && !rtnStr.trim().isEmpty();
    }

    public final static String getVmsClientFilter(Properties props) {
        String rtnStr = props.getProperty(vmsClientFilter);
        if (rtnStr == null || rtnStr.trim().isEmpty()) {
            return "*.*.*.*";
        }
        return rtnStr.trim();
    }

    public final static String getVmsClusterPath(Properties props) {
        return props.getProperty(vmsMetadataCluster);
    }

    public final static String getVmsChannelName(Properties props) {
        return props.getProperty(vmsChannelMetadataTopic);
    }

    public final static String getQueueShards(Properties props) {
        return props.getProperty(vmsRabbitmqShards);
    }

    public final static String getVmsChannelSubQueues(Properties props) {
        return props.getProperty(vmsChannelSubQueues);
    }

    public final static boolean isVmsQueueSharding(Properties props) {
        if (props.getProperty(vmsIsQueueSharding) == null)
            return false;
        try {
            return Boolean.parseBoolean(props.getProperty(vmsIsQueueSharding));
        } catch (Exception e) {
            return false;
        }
    }

    public final static String getVmsQueueName(Properties props) {
        return props.getProperty(vmsQueueMetadataGroup);
    }

    public final static String getVmsBrokerCluster(Properties props) {
        return props.getProperty(vmsClusterMetadataBroker);
    }

    public final static String getCfgZkCluster(Properties props) {
        return props.getProperty(vmsClusterMetadataCfgsvr);
    }

    public final static String getVmsBrokerType(Properties props) {
        return props.getProperty(vmsClusterMetadataBrokerType);
    }

    public final static String getVmsCfgZkType(Properties props) {
        return props.getProperty(vmsClusterMetadataCfgsvrType);
    }

    public final static String getVmsUserName(Properties props) {
        return props.getProperty(vmsMetadataUser, "root");
    }

    public final static String getVmsUserPasswd(Properties props) {
        return props.getProperty(vmsMetadataPasswd, "osroot");
    }

    final static class ZKConfig {
        private String zkConnect;
        private int zkSessionTimeoutMs;
        private int zkConnectionTimeoutMs;

        public ZKConfig(Properties props) {
            zkConnect = props.getProperty("zk.connection", "");
            try {
                zkSessionTimeoutMs = Integer.parseInt(props.getProperty("zk.session.timeout.ms"));
            } catch (NumberFormatException e) {
                zkSessionTimeoutMs = 6000;
            }
            try {
                zkConnectionTimeoutMs = Integer.parseInt(props.getProperty("zk.connection.timeout.ms"));
            } catch (NumberFormatException e) {
                zkConnectionTimeoutMs = zkSessionTimeoutMs;
            }
        }

        public String getZkConnect() {
            return zkConnect;
        }

        public int getZkSessionTimeoutMs() {
            return zkSessionTimeoutMs;
        }

        public int getZkConnectionTimeoutMs() {
            return zkConnectionTimeoutMs;
        }
    }

    public final static class ZKStrSerializer implements ZkSerializer {

        @Override
        public byte[] serialize(Object data) throws ZkMarshallingError {
            byte[] stream = null;
            try {
                stream = data instanceof String ? ((String) data).getBytes(vmsCharEncoding) : new byte[0];
            } catch (UnsupportedEncodingException e) {
                stream = new byte[0];
            }
            return stream;
        }

        @Override
        public Object deserialize(byte[] bytes) throws ZkMarshallingError {
            if (bytes == null) {
                return null;
            } else {
                try {
                    return new String(bytes, vmsCharEncoding);
                } catch (UnsupportedEncodingException e) {
                    throw new ZkMarshallingError(e);
                }
            }
        }
    }

    public static ZkClient createZkClient(String zkUrl, int sessionTimeout, int connectionTimeout) {
        return new ZkClient(zkUrl, sessionTimeout, connectionTimeout, new ZKStrSerializer());
    }

    public static ZkClient createZkClient(ZKConfig zkConfig) {
        return createZkClient(zkConfig.zkConnect, zkConfig.zkSessionTimeoutMs, zkConfig.zkConnectionTimeoutMs);
    }

    public static boolean pathExists(ZkClient zkClient, String path) {
        return zkClient.exists(path);
    }

    public static Tuple<String, Stat> readData(ZkClient zkClient, String path) {
        Stat stat = new Stat();
        String dataStr = zkClient.readData(path, stat);
        return new Tuple<String, Stat>(dataStr, stat);
    }

    public static Properties readProperties(ZkClient zkClient, String path) throws IOException {
        Properties properties = null;
        log.info("读取zk目录:\n{}",path);
        try {
            String dataStr = readData(zkClient, path).item1();
            log.info("返回:\n{}",dataStr);
            properties = Utils.parseProperties(dataStr);
        } catch (Exception e) {
            logger.error("Read zknode path of [{}] error.", path, e);
            throw new IOException("Read zknode path of [" + path + "] error.");
        }

        return properties;
    }

    public static Tuple<String, Stat> readDataMaybeNull(ZkClient zkClient, String path) {
        Stat stat = new Stat();
        String rtStr = "";
        try {
            rtStr = zkClient.readData(path, stat);
        } catch (ZkNoNodeException e) {
            return new Tuple<String, Stat>(rtStr, stat);
        }
        return new Tuple<String, Stat>(rtStr, stat);
    }

    public static void maybeDeletePath(String zkUrl, String dir) {
        try {
            ZkClient zk = createZkClient(zkUrl, 30 * 1000, 30 * 1000);
            zk.deleteRecursive(dir);
            zk.close();
        } catch (Throwable e) {
            logger.warn("The operation of maybeDeletePath be failed.", e);
        }
    }

    public static void makeSurePersistentPathExists(ZkClient zkClient, String path) throws VmsClientException {
        if (!zkClient.exists(path)) {
            ZKPath.createPersistent(zkClient, path, true);
        }
    }

    private static void createParentPath(ZkClient zkClient, String path) throws VmsClientException {
        String parentDir = path.substring(0, path.lastIndexOf('/'));
        if (!parentDir.isEmpty()) {
            ZKPath.createPersistent(zkClient, parentDir, true);
        }
    }

    private static void createEphemeralPath(ZkClient zkClient, String path, String data) {
        try {
            ZKPath.createEphemeral(zkClient, path, data);
        } catch (VmsClientException e) {
            e.printStackTrace();
        }
    }

    public static class ZKPath {
        private static volatile boolean isNamespacePresent = false;

        public static void checkNamespace(ZkClient zkClient) throws VmsClientException {
            if (isNamespacePresent)
                return;
            if (!zkClient.exists("/")) {
                throw new VmsClientException("Zk namespace [/] not exists.",
                        VmsClientErrorCode.ZK_ERR_CODE_NAMESPACE_NOT_EXISTS);
            }
            isNamespacePresent = true;
        }

        public static void resetNamespaceCheckedState() {
            isNamespacePresent = false;
        }

        public static void createPersistent(ZkClient zkClient, String path, Object data) throws VmsClientException {
            checkNamespace(zkClient);
            zkClient.createPersistent(path, data);
        }

        public static void createPersistent(ZkClient zkClient, String path, boolean createPaents)
                throws VmsClientException {
            checkNamespace(zkClient);
            zkClient.createPersistent(path, createPaents);
        }

        public static void createEphemeral(ZkClient zkClient, String path, Object data) throws VmsClientException {
            checkNamespace(zkClient);
            zkClient.createEphemeral(path, data);
        }

        public static String createEphemeralSequential(ZkClient zkClient, String path, Object data)
                throws VmsClientException {
            checkNamespace(zkClient);
            return zkClient.createEphemeralSequential(path, data);
        }

        public static void createPersistentSequential(ZkClient zkClient, String path, Object data)
                throws VmsClientException {
            checkNamespace(zkClient);
            zkClient.createPersistentSequential(path, data);
        }

        public static void delete(ZkClient zkClient, String path) {
            zkClient.delete(path);
        }

        public static void deleteRecursive(ZkClient zkClient, String path) {
            zkClient.deleteRecursive(path);
        }

        public static void registerDataChangeListener(ZkClient zkClient, String path, IZkDataListener listener) {
            zkClient.subscribeDataChanges(path, listener);
        }

        public static void unregisterDataChangeListener(ZkClient zkClient, String path, IZkDataListener listener) {
            zkClient.unsubscribeDataChanges(path, listener);
        }

        public static void registerChildChangeListener(ZkClient zkClient, String path, IZkChildListener listener) {
            zkClient.subscribeChildChanges(path, listener);
        }

        public static void unregisterChildChangeListener(ZkClient zkClient, String path, IZkChildListener listener) {
            zkClient.unsubscribeChildChanges(path, listener);
        }

        public static void registerStateChangeListener(ZkClient zkClient, IZkStateListener listener) {
            zkClient.subscribeStateChanges(listener);
        }

        public static void unregisterStateChangeListener(ZkClient zkClient, IZkStateListener listener) {
            zkClient.unsubscribeStateChanges(listener);
        }

    }

}
